#include "mqtt_msg.h"
#include "main.h"
#include <cstring>
#include <unistd.h>

namespace mqtt {
bool Mqtt::mqtt_connect_flag = false;
static int mqtt_pub_callback_id = -1;
static queue<msg_t> mqtt_msg_queue; /* mqtt消息缓冲区,队列 */

/* mqtt接收回调函数 */
void mqtt_recv_message_callback(mosquitto *mqtt_id,
                                void *obj,
                                const mosquitto_message *message)
{
    if (!message->topic || !message->payload)
        return;

    sys_log_info("topic: %s\npayload: %s\n", message->topic, message->payload);
    
    msg_t msg;

    msg.topic = (char *)message->topic;
    msg.payload = (char *)message->payload;

    mqtt_msg_queue.push(msg);
}
/* mqtt连接回调函数 */
void mqtt_connect_callback(mosquitto *mqtt_id, void *obj, int flag)
{
    if (!flag) {
        Mqtt::mqtt_connect_flag = true;
        sys_log_info("usb mqtt connect success\n");
    } else {
        Mqtt::mqtt_connect_flag = false;
    }
}
/* mqtt断开连接回调函数 */
void mqtt_disconnect_callback(mosquitto *mqtt_id, void *obj, int flag)
{
    Mqtt::mqtt_connect_flag = false;
}
/* mqtt发布回调函数 */
void mqtt_publish_callback(mosquitto *mqtt_id, void *obj, int flag)
{
    mqtt_pub_callback_id = flag;
}

/* mqtt订阅回调函数 */
void mqtt_subscribe_callback(mosquitto *mqtt_id,
                             void *obj,
                             int mid,
                             int qos_count,
                             const int *granted_qos)
{
    /* none */
}

/* 静态成员函数 */
bool Mqtt::MqttIsConnect(void)
{
    return mqtt_connect_flag;
}

void Mqtt::Mosquitto_version(void)
{
	int major, minor, revision;

	mosquitto_lib_version(&major, &minor, &revision);
	sys_log_info("mosquittom running on libmosquitto %d.%d.%d.\n", major, minor, revision);
}

Mqtt::Mqtt()
{
    /* 构造函数中,mqtt初始化 */
    mosquitto_lib_cleanup();
    mosquitto_lib_init();
    Mosquitto_version();

    int ret = MOSQ_ERR_NO_CONN;

    if (local_mqtt_id != NULL) {
        mosquitto_loop_stop(local_mqtt_id, true);
        mosquitto_destroy(local_mqtt_id);
        local_mqtt_id = NULL;
    }

    /* 申请ID */
    local_mqtt_id = mosquitto_new("modbus_usb", true, NULL);
    /* 设置接收回调 */
    mosquitto_message_callback_set(local_mqtt_id, mqtt_recv_message_callback);
    /* 设置连接回调 */
    mosquitto_connect_callback_set(local_mqtt_id, mqtt_connect_callback);
    /* 设置断开连接回调 */
    mosquitto_disconnect_callback_set(local_mqtt_id, mqtt_disconnect_callback);
    /* 设置发布回调 */
    mosquitto_publish_callback_set(local_mqtt_id, mqtt_publish_callback);
    /* 设置订阅回调 */
    mosquitto_subscribe_callback_set(local_mqtt_id, mqtt_subscribe_callback);
    /* 连接 */
    ret = mosquitto_connect(local_mqtt_id, "127.0.0.1", 1883, 60);
    if (MOSQ_ERR_SUCCESS != ret) {
        sys_log_error("Mqtt connect to error:%s\n", mosquitto_strerror(ret));
        return;
    }
    /* 开启消息循环 */
    mosquitto_loop_start(local_mqtt_id);
    while (!Mqtt::MqttIsConnect()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        sys_log_info("wait mqtt connect\n");
    }
}
/* 发布函数 */
bool Mqtt::Publish(const char *topic, const char *data)
{
    if (!topic || !data)
        return false;

    int wait = 10;
    mqtt_pub_id++;
    sys_log_info("pub topic: %s\n", topic);
    sys_log_info("pub data: %s\n", data);
    int ret = mosquitto_publish(local_mqtt_id,
                                &mqtt_pub_id,
                                topic,
                                strlen(data),
                                data,
                                MQTT_QOS,
                                false);
    if (ret != MOSQ_ERR_SUCCESS) {
        sys_log_error("Publish to %s error:%s\n", topic, mosquitto_strerror(ret));
        return false;
    }
    while (wait--) {
        if (mqtt_pub_id == mqtt_pub_callback_id) {
            return true;
        }
        usleep(10);
    }
    return false;
}
/* 订阅函数 */
bool Mqtt::Subscribe(const char *topic)
{
    if (!topic)
        return false;
    int rc = MOSQ_ERR_NO_CONN;
    sys_log_info("sub topic: %s\n", topic);
    rc = mosquitto_subscribe(local_mqtt_id, NULL, topic, MQTT_QOS);
    if (rc != MOSQ_ERR_SUCCESS) {
        sys_log_error("Subscribe to %s error:%s\n", topic, mosquitto_strerror(rc));
        return false;
    }
    return true;
}

/* 取消订阅 */
bool Mqtt::UnSubscribe(const char *topic)
{
    if (!topic)
        return false;
    int rc = MOSQ_ERR_NO_CONN;
    sys_log_info("unsub topic: %s\n", topic);
    rc = mosquitto_unsubscribe(local_mqtt_id, NULL, topic);
    if (rc != MOSQ_ERR_SUCCESS) {
        sys_log_error("UnSubscribe to %s error:%s\n", topic, mosquitto_strerror(rc));
        return false;
    }
    return true;
}

Mqtt::~Mqtt()
{
    /* none */
}
bool MqttMsgQueueIsEmpty(void)
{
    return mqtt_msg_queue.empty();
}
msg_t &MqttMsgQueueGetItem(void)
{
    return mqtt_msg_queue.front();
}
void MqttMsgQueuePop(void)
{
    mqtt_msg_queue.pop();
}
int MqttMsgQueueGetCount(void)
{
    return mqtt_msg_queue.size();
}
}